#ifndef __M_CONNECTION_H__
#define __M_CONNECTION_H__
#include "../thirdLib/proto/codec.h"
#include "../thirdLib/proto/dispatcher.h"
#include "../thirdLib/base/Mutex.h"
#include "../thirdLib/net/EventLoop.h"
#include "../thirdLib/net/TcpClient.h"
#include "../thirdLib/net/EventLoopThread.h"
#include "../thirdLib/base/CountDownLatch.h"
#include "../mqCommon/logger.hpp"
#include "../mqCommon/helper.hpp"
#include "../mqCommon/mq_thread_pool.hpp"
#include "../mqCommon/mq_msg.pb.h"
#include "../mqCommon/mq_proto.pb.h"
#include "mq_channel.hpp"
#include "mq_worker.hpp"

namespace yqx
{
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
        Connection(const std::string &sip, int sport, const AsyncWorker::ptr& worker) : 
        _latch(1),
        _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
        _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
        _client(worker->loopthread.startLoop(), muduo::net::InetAddress(sip, sport), "Client"),
        _worker(worker),
        _channel_manager(std::make_shared<ChannelManager>())
        {
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::basicResponse, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::consumeResponse, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));

            _client.connect();
            _latch.wait();//阻塞等待，直到连接成功
        }

        Channel::ptr openChannel(){
            Channel::ptr channel = _channel_manager->create(_conn, _codec);
            bool ret = channel->openChannel();
            if(ret == false){
                LOG("打开信道失败!");
                return nullptr;
            }
            return channel;
        }
        void closeChannel(const Channel::ptr& channel){
            channel->closeChannel();
            _channel_manager->remove(channel->cid());
        }

    private:
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            LOG("unknowMessage!");
            conn->shutdown();
        }

        void basicResponse(const muduo::net::TcpConnectionPtr &conn, const basicCommonResponsePtr &message, muduo::Timestamp)
        {
            //找到信道
            Channel::ptr channel = _channel_manager->getChannel(message->cid());
            if(channel == nullptr){
                LOG("未找到信道!");
                return;
            }
            //将得到的响应对象，添加到信道的基础响应hash_map中
            channel->putBasicResponse(message);
        }

        void consumeResponse(const muduo::net::TcpConnectionPtr &conn, const basicConsumeResponsePtr &message, muduo::Timestamp)
        {
            //找到信道
            Channel::ptr channel = _channel_manager->getChannel(message->cid());
            if(channel == nullptr){
                LOG("未找到信道!");
                return;
            }
            //封装异步任务（消息处理任务），抛入线程池
            _worker->pool.push([channel, message](){
                channel->consume(message);
            });
        }

        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                _latch.countDown();
                _conn = conn;
                std::cout << "Connect succed!" << std::endl;
            }
        }

    private:
        muduo::CountDownLatch _latch;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;
        muduo::net::TcpClient _client;
        AsyncWorker::ptr _worker;
        ChannelManager::ptr _channel_manager;
    };
}

#endif